package com.zhang.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import com.alibaba.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.alibaba.ververica.cdc.debezium.DebeziumSourceFunction;
import com.zhang.gmall.realtime.app.func.MyDeserializationSchema;
import com.zhang.gmall.realtime.beans.TableProcess;
import com.zhang.gmall.realtime.common.GmallConfig;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.sql.DriverManager;

/**
 * 重点
 *
 * @title: 业务数据动态分流
 * @author: zhang
 * @date: 2022/3/3 09:28
 * 事实数据：放到kafka不同主题
 * 维度数据：放到Hbase（通过Phoenix）不同表
 */
public class BaseDBApp_CDC_test {
    public static void main(String[] args) throws Exception {
        //todo 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        //todo 2.设置检查点

        DebeziumSourceFunction<String> sourceFunction = MySQLSource.<String>builder()
                .hostname("hadoop103")
                .port(3306)
                .username("root")
                .password("000000")
                .databaseList("gmall_realtime_2022")
                .tableList("gmall_realtime_2022.table_process")
                .startupOptions(StartupOptions.initial())
                .deserializer(new MyDeserializationSchema())
                .build();
        DataStreamSource<String> tableProcessStrDS = env.addSource(sourceFunction);
       tableProcessStrDS.process(new ProcessFunction<String, String>() {
           private Connection conn;
           @Override
           public void open(Configuration parameters) throws Exception {
               //注册驱动
               Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
               //创建连接
               conn = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
           }

           @Override
           public void processElement(String value, ProcessFunction<String, String>.Context ctx, Collector<String> out) throws Exception {
               //获取并解析数据
               JSONObject jsonObject = JSON.parseObject(value);
               String data = jsonObject.getString("data");
               TableProcess tableProcess = JSON.parseObject(data, TableProcess.class);

               if (TableProcess.SINK_TYPE_HBASE.equals(tableProcess.getSinkType())) {
                   checkTable(
                           tableProcess.getSinkTable(),
                           tableProcess.getSinkColumns(),
                           tableProcess.getSinkPk(),
                           tableProcess.getSinkExtend()
                   );
               }
           }
       }).print();

        //todo 11.执行任务
        env.execute("BaseDBApp");
    }

    private static void checkTable(String sinkTable, String sinkColumns, String sinkPk, String sinkExtend) {

            if (sinkPk == null) {
                sinkPk = "id";
            }
            if (sinkExtend == null) {
                sinkExtend = "";
            }

            StringBuffer creatTableSQL = new StringBuffer("CREATE TABLE IF NOT EXISTS ")
                    .append(GmallConfig.HBASE_SCHEMA)
                    .append(".")
                    .append(sinkTable)
                    .append("(");
            String[] fields = sinkColumns.split(",");

            for (int i = 0; i < fields.length; i++) {
                String field = fields[i];
                //判断是否为主键
                if (field.equals(sinkPk)) {
                    creatTableSQL.append(field).append(" VARCHAR primary key ");
                } else {
                    creatTableSQL.append(field).append(" VARCHAR ");
                }
                if (i < fields.length - 1) {
                    creatTableSQL.append(",");
                }
            }

            creatTableSQL.append(")").append(sinkExtend);
            System.out.println(creatTableSQL);


    }
}
